package com.fanxuankai.boot.canal.client;

import com.fanxuankai.boot.canal.client.autoconfigure.CanalClientProperties;
import com.fanxuankai.commons.util.concurrent.Flow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Canal 事务确认订阅者
 *
 * @author fanxuankai
 */
public class ConfirmSubscriber implements Flow.Subscriber<MessageWrapper> {
    private static final Logger LOGGER = LoggerFactory.getLogger(ConfirmSubscriber.class);
    private final OtterClient otterClient;
    private final CanalClientProperties properties;
    private Flow.Subscription subscription;

    public ConfirmSubscriber(OtterClient otterClient, CanalClientProperties properties) {
        this.otterClient = otterClient;
        this.properties = properties;
    }

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        subscription.request(1);
    }

    @Override
    public void onNext(MessageWrapper item) {
        long start = System.currentTimeMillis();
        otterClient.ack(item.getBatchId());
        long t = System.currentTimeMillis() - start;
        if (properties.isShowEventLog() && !item.getEntryWrapperList().isEmpty()) {
            LOGGER.info("[" + properties.getId() + "] " + "Confirm batchId: {} time: {}ms", item.getBatchId()
                    , t);
        }
        subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        LOGGER.error("[" + properties.getId() + "] " + throwable.getLocalizedMessage(), throwable);
        this.subscription.cancel();
        this.otterClient.stop();
    }

    @Override
    public void onComplete() {
        LOGGER.info("[" + properties.getId() + "] " + "Done");
    }
}
